
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class apitest {

    private static final String tableName = "ci:student";
    private static final String nameSpace = "ci";
    private static final String colFamily1 = "info";
    private static final String colFamily2 = "score";
    private static final String rowkey1 = "Tom";
    private static final String rowkey2 = "Jerry";
    private static final String rowkey3 = "Jack";
    private static final String rowkey4 = "Rose";
    private static final String rowkey5 = "ci";
    private static final String col1 = "student_id";
    private static final String col2 = "class";
    private static final String col3 = "understanding";
    private static final String col4 = "programming";
    private static Connection connection;


    /**
     * 判断表是否存在
     *
     * @param tableName
     * @return
     */
    public static boolean tableIfExists(String tableName) {
        try {
            HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;


    }


    /**
     * 创建 HBase 表
     *
     * @param tableName      表名
     * @param columnFamilies 列族的数组
     */
    public static boolean createTable(String tableName, String nameSpace, List<String> columnFamilies) {
        try {
            HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();

            TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
            NamespaceDescriptor descriptor = NamespaceDescriptor.create(nameSpace).build();

            columnFamilies.forEach(columnFamily -> {
                ColumnFamilyDescriptorBuilder cfDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));
                cfDescriptorBuilder.setMaxVersions(1);
                ColumnFamilyDescriptor familyDescriptor = cfDescriptorBuilder.build();
                tableDescriptor.setColumnFamily(familyDescriptor);
            });
            admin.createNamespace(descriptor);
            admin.createTable(tableDescriptor.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 根据 rowkey 获得指定行数据
     *
     * @param tableName
     * @param rowKey
     * @return
     */
    public static Result getRow(String tableName, String rowKey) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            return table.get(get);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 删除 hBase 表
     *
     * @param tableName 表名
     */
    public static boolean deleteTable(String tableName, String nameSpace) {
        try {
            HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();

            // 删除表前需要先禁用表
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 插入数据
     *
     * @param tableName        表名
     * @param rowKey           唯一标识
     * @param columnFamilyName 列族名
     * @param qualifier        列标识
     * @param value            数据
     */
    public static boolean putRow(String tableName, String rowKey, String columnFamilyName, String qualifier,
                                 String value) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            table.put(put);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }
    /**
     * 删除数据
     *
     * @param tableName        表名
     * @param rowKey           唯一标识
     */
    public static boolean deleteRow(String tableName, String rowKey) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    public static void main(String[] args) throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "jikehadoop01");

        try {
            connection = ConnectionFactory.createConnection(configuration);
            System.out.println("connected");
        } catch (IOException e) {
            e.printStackTrace();
        }
        deleteRow(tableName,rowkey1);
        System.out.println("getRow: " + getRow(tableName, rowkey1));

//        if (tableIfExists(tableName)) {
//            System.out.println("table already exists");
//            deleteTable(tableName, nameSpace);
//        } else {
//            List<String> colFamilyList = new ArrayList<>();
//            colFamilyList.add(colFamily1);
//            colFamilyList.add(colFamily2);
//            System.out.println(colFamilyList);
//            System.out.println("start create table");
//            createTable(tableName, nameSpace, colFamilyList);
//
//            putRow(tableName, rowkey1, colFamily1, col1, "20210000000001");
//            putRow(tableName, rowkey1, colFamily1, col2, "1");
//            putRow(tableName, rowkey1, colFamily2, col3, "75");
//            putRow(tableName, rowkey1, colFamily2, col4, "82");
//
//            putRow(tableName, rowkey2, colFamily1, col1, "20210000000002");
//            putRow(tableName, rowkey2, colFamily1, col2, "1");
//            putRow(tableName, rowkey2, colFamily2, col3, "85");
//            putRow(tableName, rowkey2, colFamily2, col4, "67");
//
//            putRow(tableName, rowkey3, colFamily1, col1, "20210000000003");
//            putRow(tableName, rowkey3, colFamily1, col2, "2");
//            putRow(tableName, rowkey3, colFamily2, col3, "80");
//            putRow(tableName, rowkey3, colFamily2, col4, "80");
//
//            putRow(tableName, rowkey4, colFamily1, col1, "20210000000004");
//            putRow(tableName, rowkey4, colFamily1, col2, "2");
//            putRow(tableName, rowkey4, colFamily2, col3, "60");
//            putRow(tableName, rowkey4, colFamily2, col4, "61");
//
//            putRow(tableName, rowkey5, colFamily1, col1, "G20200389020012");
//            putRow(tableName, rowkey5, colFamily1, col2, "2");
//            putRow(tableName, rowkey5, colFamily2, col3, "90");
//            putRow(tableName, rowkey5, colFamily2, col4, "90");
//
//            System.out.println("getRow: " + getRow(tableName, rowkey1));
//            System.out.println("getRow: " + getRow(tableName, rowkey2));
//            System.out.println("getRow: " + getRow(tableName, rowkey3));
//            System.out.println("getRow: " + getRow(tableName, rowkey4));
//            System.out.println("getRow: " + getRow(tableName, rowkey5));
//        }
    }


}